package com.xy.springboot_3.reactor.demo;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.publisher.Flux;

import java.time.Duration;

@Slf4j
public class ApiReactorDemo {

    /**
     * transform 将流进行传输
     */
    @Test
    public void transformDemo(){

        Flux<Object> transform = Flux.just(1,2,3)
                .transformDeferred(values ->{
                    return values.map(item->item);
                })
                .transform(values->{


                   return values.map(item -> "a"+item);
                });

    }

    /**
     * concat、 对流进行拼接
     * concatWith
     */
    @Test
    public void concatDemo(){
        Flux.concat(Flux.just(1,2,3),Flux.just("4",5,"6"))
                .subscribe(System.out::println);
        /**
         * 1
         * 2
         * 3
         * 4
         * 5
         * 6
         */
    }

    /**
     * merge、      静态方法，能把 同类型的 流组合成新得流
     * mergeWith、   流操作过程种再次组合
     */

    @Test
    public void mergeDemo(){
        Flux.merge(Flux.just(1,2,3,4,5),Flux.just(6,7,8))
                .mergeWith(Flux.just(9,10))
                //.log()
                .subscribe(System.out::println);

        /**
         * 1
         * 2
         * 3
         * 4
         * 5
         * 6
         * 7
         * 8
         * 9
         * 10
         */
    }




    /**
     * zip、 进行压缩元素 组成新元祖数据  静态方法
     * zipWith...   进行压缩元素 组成新元祖数据 流中的组合方法
     */
    @Test
    public void zipTest(){

        Flux.zip(Flux.just(1,2,3),Flux.just("a","b","c"))
                .map(item -> {
                    log.info(item.getT1()+"");
                    log.info(item.getT2());

                    return Thread.currentThread().getName()+":"+ item.getT1() + item.getT2();
                })
                .subscribe(System.out::println);
        /**
         * 21:15:44.838 [main] INFO com.xy.springboot_3.reactor.demo.ApiReactorDemo -- 1
         * 21:15:44.845 [main] INFO com.xy.springboot_3.reactor.demo.ApiReactorDemo -- a
         * main:1a
         * 21:15:44.849 [main] INFO com.xy.springboot_3.reactor.demo.ApiReactorDemo -- 2
         * 21:15:44.850 [main] INFO com.xy.springboot_3.reactor.demo.ApiReactorDemo -- b
         * main:2b
         * 21:15:44.850 [main] INFO com.xy.springboot_3.reactor.demo.ApiReactorDemo -- 3
         * 21:15:44.850 [main] INFO com.xy.springboot_3.reactor.demo.ApiReactorDemo -- c
         * main:3c
         */


 // 如果不对应 则 忽略 后续 zip 数据，
        Flux.zip(Flux.just(1,2,3),Flux.just("a"))
                .map(item -> {
                    log.info(item.getT1()+"");
                    log.info(item.getT2());

                    return Thread.currentThread().getName()+":"+ item.getT1() + item.getT2();
                })
                .subscribe(System.out::println);

        /**
         * 21:16:23.161 [main] INFO com.xy.springboot_3.reactor.demo.ApiReactorDemo -- 1
         * 21:16:23.162 [main] INFO com.xy.springboot_3.reactor.demo.ApiReactorDemo -- a
         * main:1a
         * Disconnected from the target VM, address: '127.0.0.1:58783', transport: 'socket'
         */

    }

    /**
     * filter 过滤 api 操作
     */
    @Test
    public void filter(){
        Flux.range(1,10)
                .filter(item -> item>5)
                //.log()
                .subscribe(System.out::println);
        /**
         * 6
         * 7
         * 8
         * 9
         * 10
         */
    }
}
